import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.UserRegisterBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 *从 DWD层用户注册表中读取数据，统计各窗口注册用户数，写入 Doris。
 */
public class DwsUserUserRegisterWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsUserUserRegisterWindow().start(10025, 4, "dws_user_user_register_window", Constant.TOPIC_DWD_USER_REGISTER);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 读取数据
        // 将流转换格式为JSONOBJECT
        SingleOutputStreamOperator<JSONObject> mapDs = stream.map(JSON::parseObject);
        // mapDs.print("MAP");

        // 设置水位线
        SingleOutputStreamOperator<UserRegisterBean>  aggregateDs = mapDs.assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3L))
                                .withIdleness(Duration.ofSeconds(120L))
                                .withTimestampAssigner((jsonObj, ts) -> jsonObj.getLong("create_time"))  // 这里fastjson会自动把datetime转换成Long
                )
                // 开窗 聚合
                .windowAll(
                        TumblingEventTimeWindows.of(Time.seconds(5L))
                ).aggregate(
                        new AggregateFunction<JSONObject, Long, Long>() {
                            @Override
                            public Long createAccumulator() {
                                return 0L;
                            }

                            @Override
                            public Long add(JSONObject jsonObject, Long aLong) {
                                return aLong + 1;
                            }

                            @Override
                            public Long getResult(Long aLong) {
                                return aLong;
                            }

                            @Override
                            public Long merge(Long aLong, Long acc1) {
                                return aLong + acc1;
                            }
                        },
                        new ProcessAllWindowFunction<Long, UserRegisterBean, TimeWindow>() {
                            @Override
                            public void process(ProcessAllWindowFunction<Long, UserRegisterBean, TimeWindow>.Context context, Iterable<Long> iterable, Collector<UserRegisterBean> collector) throws Exception {
                                Long result = iterable.iterator().next();
                                Long stt = context.window().getStart();
                                Long edt = context.window().getEnd();
                                Long curDate = context.window().getStart();
                                collector.collect(
                                        UserRegisterBean.builder()
                                                .stt(DateFormatUtil.tsToDateTime(stt))
                                                .edt(DateFormatUtil.tsToDateTime(edt))
                                                .curDate(DateFormatUtil.tsToDate(stt))
                                                .registerCt(result)
                                                .build()
                                );
                            }
                        }
                );

        // aggregateDs.print("AGGREGATE");


        // 写如Doris
        aggregateDs.map(
                new BeanToJsonMapFunction<>()
        ).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DORIS_DATABASE, Constant.DWS_USER_USER_REGISTER_WINDOW));


    }
}
